package com.zyx.flinkdemo.stream.window.join;

import com.zyx.flinkdemo.pojo.ClickBean;
import com.zyx.flinkdemo.pojo.Trade;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.CoGroupedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Iterator;
import java.util.List;

/**
 * @author Yaxi.Zhang
 * @since 2021/6/30 17:53
 */
public class TumblingWindowCoGroupTemplate {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        List<ClickBean> clicksData = PrepareData.getClicksData();
        DataStream<ClickBean> clickStream = env.fromCollection(clicksData)
                .assignTimestampsAndWatermarks(((WatermarkStrategy<ClickBean>) context -> new WatermarkGenerator<ClickBean>() {
                    @Override
                    public void onEvent(ClickBean clickBean, long eventTimestamp, WatermarkOutput output) {
                        output.emitWatermark(new Watermark(clickBean.getVisitTime().getTime()));
                    }

                    @Override
                    public void onPeriodicEmit(WatermarkOutput output) {

                    }
                })
                        .withTimestampAssigner((SerializableTimestampAssigner<ClickBean>) (clickBean, recordTimestamp) -> clickBean.getVisitTime().getTime()));

        List<Trade> personData = PrepareData.getTradeData();
        DataStream<Trade> tradeStream = env
                .fromCollection(personData)
                .assignTimestampsAndWatermarks(((WatermarkStrategy<Trade>) context -> new WatermarkGenerator<Trade>() {
                    @Override
                    public void onEvent(Trade trade, long eventTimestamp, WatermarkOutput output) {
                        output.emitWatermark(new Watermark(trade.getTradeTime().getTime() - 1));
                    }

                    @Override
                    public void onPeriodicEmit(WatermarkOutput output) {

                    }
                })
                        .withTimestampAssigner((SerializableTimestampAssigner<Trade>) (trade, recordTimestamp) -> trade.getTradeTime().getTime()));


        CoGroupedStreams.WithWindow<ClickBean, Trade, String, TimeWindow> window =
                clickStream.coGroup(tradeStream)
                        .where((KeySelector<ClickBean, String>) ClickBean::getUser)
                        .equalTo((KeySelector<Trade, String>) Trade::getName)
                        .window(TumblingEventTimeWindows.of(Time.hours(2)));


        DataStream<String> apply = window.apply(new CoGroupFunction<ClickBean, Trade, String>() {
            @Override
            public void coGroup(Iterable<ClickBean> clickStream,
                                Iterable<Trade> tradeStream, Collector<String> out) {
                for (ClickBean clickBean : clickStream) {
                    Iterator<Trade> iterator = tradeStream.iterator();
                    StringBuilder mess = new StringBuilder();
                    while (iterator.hasNext()) {
                        Trade next = iterator.next();
                        mess.append("--").append(next.getName()).append("--").append(next.getClient());
                    }
                    out.collect(clickBean.getUser() + ":" + clickBean.getUrl() + ":" + mess);
                }
            }
        });
        apply.print("CoGroup输出结果");
        env.execute("WindowCoGroupSlidTemplate");

    }
}
